package cn.doitedu.sql;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/5
 * @Desc: 学大数据，上多易教育
 * 表  转  流
 **/
public class _24_TableToStream {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        tenv.executeSql(
                " create table score_kfk(                                      " +
                        "       user_id int,                                    " +
                        "       course string,                                  " +
                        " 	    score double                                     " +
                        "  ) with (                                              " +
                        "       'connector' = 'kafka',                           " +
                        "       'topic' = 'score-test',                            " +
                        "       'properties.bootstrap.servers' = 'doitedu:9092', " +
                        "       'properties.group.id' = 'g003',                  " +
                        "       'scan.startup.mode' = 'latest-offset',           " +
                        "       'value.format' = 'json',                         " +
                        "       'value.fields-include' = 'EXCEPT_KEY'            " +
                        " )                                                      "
        );


        /**
         * 一、 appendOnly 表专流
         */
        Table table = tenv.from("score_kfk");  // 先把 带名字的表，转成  表对象

        // 表转流时，指定目标java类型，就自动转成目标类型的流
        DataStream<MyBean> dataStream1 = tenv.toDataStream(table, MyBean.class);


        // 表转流时，如果不指定目标类型，则得到Row类型的流
        DataStream<Row> dataStream2 = tenv.toDataStream(table);
        dataStream2.map(new MapFunction<Row, String>() {
            @Override
            public String map(Row row) throws Exception {
                // 从row中获取字段数据
                int user_id = row.getFieldAs("user_id");
                String course = row.getFieldAs("course");

                return null;
            }
        });


        /**
         * 二、 changelog 表专流
         */
        // 如下查询就会得到一个  changelog 的表： sum_res
        tenv.executeSql(
                "create temporary view sum_res as " +
                "select course,sum(score) as score_sum from score_kfk group by course");
        // changelog表转流
        Table t2 = tenv.from("sum_res");
        DataStream<Row> changelogStream = tenv.toChangelogStream(t2);
        changelogStream.process(new ProcessFunction<Row, String>() {
            @Override
            public void processElement(Row row, ProcessFunction<Row, String>.Context ctx, Collector<String> out) throws Exception {

                String course = row.getFieldAs("course");
                long score_sum = row.getFieldAs("score_sum");

                // 取本行的 操作类型： -U/+U/+I/-D
                RowKind rowKind = row.getKind();

                if(rowKind.shortString().equals("+I")){
                    // TODO
                }

                if(rowKind == RowKind.UPDATE_BEFORE){  // 等价： rowKind.shortString().equals("-U")
                    // TODO
                }

            }
        });


    }



    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyBean{
        private int user_id;
        private String course;
        private double score;
    }



}
